"""
OpenGauss 数据库后端测试用例
验证 Django ORM 与 OpenGauss 的完整集成功能
"""
import unittest
import json
from decimal import Decimal
from datetime import datetime, date, time
from django.test import TestCase, TransactionTestCase
from django.db import models, connection, transaction
from django.db.models import Q, F, Count, Sum, Avg
from django.core.management import call_command
from django.test.utils import override_settings
from django.conf import settings


# 测试模型定义
class TestProduct(models.Model):
    """测试商品模型"""
    name = models.CharField(max_length=100, verbose_name="商品名称")
    description = models.TextField(verbose_name="商品描述")
    price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name="价格")
    quantity = models.IntegerField(default=0, verbose_name="库存数量")
    is_active = models.BooleanField(default=True, verbose_name="是否启用")
    created_at = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
    updated_at = models.DateTimeField(auto_now=True, verbose_name="更新时间")
    metadata = models.JSONField(default=dict, verbose_name="元数据")
    
    class Meta:
        db_table = 'test_product'
        indexes = [
            models.Index(fields=['name']),
            models.Index(fields=['price']),
            models.Index(fields=['created_at']),
        ]
    
    def __str__(self):
        return self.name


class TestCategory(models.Model):
    """测试分类模型"""
    name = models.CharField(max_length=50, unique=True, verbose_name="分类名称")
    parent = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True, verbose_name="父分类")
    level = models.SmallIntegerField(default=1, verbose_name="层级")
    
    class Meta:
        db_table = 'test_category'


class TestProductCategory(models.Model):
    """商品分类关联模型"""
    product = models.ForeignKey(TestProduct, on_delete=models.CASCADE)
    category = models.ForeignKey(TestCategory, on_delete=models.CASCADE)
    is_primary = models.BooleanField(default=False, verbose_name="是否主分类")
    
    class Meta:
        db_table = 'test_product_category'
        unique_together = ['product', 'category']


class OpenGaussBasicTest(TestCase):
    """OpenGauss 基础功能测试"""
    
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        # 确保表已创建
        with connection.schema_editor() as schema_editor:
            schema_editor.create_model(TestProduct)
            schema_editor.create_model(TestCategory)
            schema_editor.create_model(TestProductCategory)
    
    def test_database_connection(self):
        """测试数据库连接"""
        self.assertEqual(connection.vendor, 'opengauss')
        self.assertEqual(connection.display_name, 'OpenGauss')
        
        # 测试基本查询
        with connection.cursor() as cursor:
            cursor.execute("SELECT 1 as test")
            result = cursor.fetchone()
            self.assertEqual(result[0], 1)
    
    def test_version_info(self):
        """测试版本信息获取"""
        with connection.cursor() as cursor:
            cursor.execute("SELECT version()")
            version = cursor.fetchone()[0]
            self.assertIn('openGauss', version)
    
    def test_schema_operations(self):
        """测试模式操作"""
        # 测试表是否创建成功
        with connection.cursor() as cursor:
            cursor.execute("""
                SELECT table_name FROM information_schema.tables 
                WHERE table_schema = 'public' AND table_name = %s
            """, ['test_product'])
            result = cursor.fetchone()
            self.assertIsNotNone(result)


class OpenGaussCRUDTest(TestCase):
    """CRUD 操作测试"""
    
    def setUp(self):
        """设置测试数据"""
        self.category = TestCategory.objects.create(
            name="电子产品",
            level=1
        )
        
        self.product = TestProduct.objects.create(
            name="iPhone 15",
            description="苹果最新款手机",
            price=Decimal('7999.00'),
            quantity=100,
            metadata={
                "brand": "Apple",
                "model": "iPhone 15",
                "storage": "128GB"
            }
        )
    
    def test_create_operation(self):
        """测试创建操作"""
        product = TestProduct.objects.create(
            name="MacBook Pro",
            description="苹果笔记本电脑",
            price=Decimal('15999.00'),
            quantity=50,
            metadata={"brand": "Apple", "screen": "14寸"}
        )
        
        self.assertEqual(product.name, "MacBook Pro")
        self.assertEqual(product.price, Decimal('15999.00'))
        self.assertEqual(product.metadata["brand"], "Apple")
        self.assertIsNotNone(product.id)
        self.assertIsNotNone(product.created_at)
    
    def test_read_operation(self):
        """测试读取操作"""
        # 通过主键查询
        product = TestProduct.objects.get(id=self.product.id)
        self.assertEqual(product.name, "iPhone 15")
        
        # 通过字段查询
        product = TestProduct.objects.get(name="iPhone 15")
        self.assertEqual(product.price, Decimal('7999.00'))
        
        # 查询所有
        count = TestProduct.objects.count()
        self.assertEqual(count, 1)
    
    def test_update_operation(self):
        """测试更新操作"""
        # 单个对象更新
        self.product.quantity = 80
        self.product.save()
        
        updated_product = TestProduct.objects.get(id=self.product.id)
        self.assertEqual(updated_product.quantity, 80)
        
        # 批量更新
        TestProduct.objects.filter(name="iPhone 15").update(price=Decimal('7599.00'))
        
        updated_product = TestProduct.objects.get(id=self.product.id)
        self.assertEqual(updated_product.price, Decimal('7599.00'))
    
    def test_delete_operation(self):
        """测试删除操作"""
        product_id = self.product.id
        
        # 单个删除
        self.product.delete()
        
        with self.assertRaises(TestProduct.DoesNotExist):
            TestProduct.objects.get(id=product_id)
        
        # 批量删除
        TestProduct.objects.create(name="测试商品1", description="测试", price=100)
        TestProduct.objects.create(name="测试商品2", description="测试", price=200)
        
        deleted_count = TestProduct.objects.filter(name__startswith="测试").delete()[0]
        self.assertEqual(deleted_count, 2)


class OpenGaussQueryTest(TestCase):
    """复杂查询测试"""
    
    def setUp(self):
        """创建测试数据"""
        # 创建分类
        self.electronics = TestCategory.objects.create(name="电子产品", level=1)
        self.phones = TestCategory.objects.create(name="手机", parent=self.electronics, level=2)
        self.laptops = TestCategory.objects.create(name="笔记本", parent=self.electronics, level=2)
        
        # 创建商品
        self.iphone = TestProduct.objects.create(
            name="iPhone 15",
            description="苹果手机",
            price=Decimal('7999.00'),
            quantity=100,
            metadata={"brand": "Apple", "type": "phone"}
        )
        
        self.macbook = TestProduct.objects.create(
            name="MacBook Pro",
            description="苹果笔记本",
            price=Decimal('15999.00'),
            quantity=50,
            metadata={"brand": "Apple", "type": "laptop"}
        )
        
        self.samsung = TestProduct.objects.create(
            name="Galaxy S24",
            description="三星手机",
            price=Decimal('6999.00'),
            quantity=80,
            is_active=False,
            metadata={"brand": "Samsung", "type": "phone"}
        )
    
    def test_filtering(self):
        """测试过滤查询"""
        # 简单过滤
        apple_products = TestProduct.objects.filter(metadata__brand="Apple")
        self.assertEqual(apple_products.count(), 2)
        
        # 价格范围过滤
        expensive_products = TestProduct.objects.filter(price__gte=10000)
        self.assertEqual(expensive_products.count(), 1)
        self.assertEqual(expensive_products.first().name, "MacBook Pro")
        
        # JSON 字段过滤
        phones = TestProduct.objects.filter(metadata__type="phone")
        self.assertEqual(phones.count(), 2)
        
        # 复合条件
        active_apple_phones = TestProduct.objects.filter(
            metadata__brand="Apple",
            metadata__type="phone",
            is_active=True
        )
        self.assertEqual(active_apple_phones.count(), 1)
    
    def test_complex_queries(self):
        """测试复杂查询"""
        # Q 对象查询
        q_result = TestProduct.objects.filter(
            Q(metadata__brand="Apple") | Q(price__lt=7000)
        )
        self.assertEqual(q_result.count(), 2)
        
        # F 对象查询
        TestProduct.objects.filter(quantity__gt=F('price') / 100).exists()
        
        # 排序
        sorted_products = TestProduct.objects.order_by('-price')
        self.assertEqual(sorted_products.first().name, "MacBook Pro")
        
        # 限制数量
        top_2 = TestProduct.objects.order_by('-price')[:2]
        self.assertEqual(len(top_2), 2)
    
    def test_aggregation(self):
        """测试聚合查询"""
        # 计数
        total_count = TestProduct.objects.count()
        self.assertEqual(total_count, 3)
        
        # 求和
        total_value = TestProduct.objects.aggregate(
            total_value=Sum(F('price') * F('quantity'))
        )['total_value']
        expected = (7999 * 100) + (15999 * 50) + (6999 * 80)
        self.assertEqual(total_value, expected)
        
        # 平均值
        avg_price = TestProduct.objects.aggregate(avg_price=Avg('price'))['avg_price']
        expected_avg = (7999 + 15999 + 6999) / 3
        self.assertAlmostEqual(float(avg_price), expected_avg, places=2)
        
        # 按条件聚合
        brand_stats = TestProduct.objects.values('metadata__brand').annotate(
            count=Count('id'),
            total_quantity=Sum('quantity')
        )
        
        apple_stats = next(s for s in brand_stats if s['metadata__brand'] == 'Apple')
        self.assertEqual(apple_stats['count'], 2)
        self.assertEqual(apple_stats['total_quantity'], 150)


class OpenGaussJSONTest(TestCase):
    """JSON 字段测试"""
    
    def test_json_operations(self):
        """测试 JSON 字段操作"""
        product = TestProduct.objects.create(
            name="测试商品",
            description="JSON 测试",
            price=100,
            metadata={
                "specifications": {
                    "color": "black",
                    "size": "large",
                    "features": ["waterproof", "wireless"]
                },
                "reviews": [
                    {"rating": 5, "comment": "很好"},
                    {"rating": 4, "comment": "不错"}
                ]
            }
        )
        
        # JSON 键查询
        black_products = TestProduct.objects.filter(
            metadata__specifications__color="black"
        )
        self.assertEqual(black_products.count(), 1)
        
        # JSON 数组查询
        wireless_products = TestProduct.objects.filter(
            metadata__specifications__features__contains="wireless"
        )
        self.assertEqual(wireless_products.count(), 1)
        
        # 更新 JSON 字段
        product.metadata["specifications"]["color"] = "white"
        product.save()
        
        updated_product = TestProduct.objects.get(id=product.id)
        self.assertEqual(
            updated_product.metadata["specifications"]["color"], 
            "white"
        )


class OpenGaussTransactionTest(TransactionTestCase):
    """事务处理测试"""
    
    def test_transaction_commit(self):
        """测试事务提交"""
        with transaction.atomic():
            product1 = TestProduct.objects.create(
                name="事务测试1",
                description="测试",
                price=100
            )
            product2 = TestProduct.objects.create(
                name="事务测试2", 
                description="测试",
                price=200
            )
        
        # 验证数据已提交
        self.assertEqual(TestProduct.objects.filter(name__startswith="事务测试").count(), 2)
    
    def test_transaction_rollback(self):
        """测试事务回滚"""
        initial_count = TestProduct.objects.count()
        
        try:
            with transaction.atomic():
                TestProduct.objects.create(
                    name="回滚测试1",
                    description="测试",
                    price=100
                )
                # 故意触发异常
                TestProduct.objects.create(
                    name="A" * 150,  # 超出字段长度限制
                    description="测试",
                    price=200
                )
        except Exception:
            pass
        
        # 验证事务已回滚
        final_count = TestProduct.objects.count()
        self.assertEqual(final_count, initial_count)
    
    def test_savepoint(self):
        """测试保存点"""
        with transaction.atomic():
            product1 = TestProduct.objects.create(
                name="保存点测试1",
                description="测试", 
                price=100
            )
            
            # 创建保存点
            savepoint = transaction.savepoint()
            
            try:
                # 这部分会回滚到保存点
                TestProduct.objects.create(
                    name="A" * 150,  # 触发错误
                    description="测试",
                    price=200
                )
            except Exception:
                transaction.savepoint_rollback(savepoint)
            
            # 继续事务
            product2 = TestProduct.objects.create(
                name="保存点测试2",
                description="测试",
                price=300
            )
        
        # 验证只有 product1 和 product2 被保存
        products = TestProduct.objects.filter(name__startswith="保存点测试")
        self.assertEqual(products.count(), 2)


class OpenGaussPerformanceTest(TestCase):
    """性能测试"""
    
    def test_bulk_operations(self):
        """测试批量操作性能"""
        import time
        
        # 批量创建
        start_time = time.time()
        products = []
        for i in range(1000):
            products.append(TestProduct(
                name=f"批量商品{i}",
                description=f"描述{i}",
                price=Decimal(str(100 + i)),
                quantity=i
            ))
        
        TestProduct.objects.bulk_create(products, batch_size=100)
        bulk_time = time.time() - start_time
        
        # 验证数据
        self.assertEqual(TestProduct.objects.filter(name__startswith="批量商品").count(), 1000)
        
        # 批量更新
        start_time = time.time()
        TestProduct.objects.filter(name__startswith="批量商品").update(is_active=False)
        update_time = time.time() - start_time
        
        # 验证更新
        inactive_count = TestProduct.objects.filter(
            name__startswith="批量商品",
            is_active=False
        ).count()
        self.assertEqual(inactive_count, 1000)
        
        print(f"批量创建1000条记录耗时: {bulk_time:.3f}秒")
        print(f"批量更新1000条记录耗时: {update_time:.3f}秒")


def run_opengauss_tests():
    """运行所有 OpenGauss 测试"""
    import sys
    from io import StringIO
    from django.test.utils import get_runner
    from django.conf import settings
    
    # 重定向输出
    old_stdout = sys.stdout
    sys.stdout = mystdout = StringIO()
    
    try:
        # 配置测试运行器
        TestRunner = get_runner(settings)
        test_runner = TestRunner(verbosity=2, interactive=False)
        
        # 运行测试
        test_labels = [
            'django_opengauss.tests.OpenGaussBasicTest',
            'django_opengauss.tests.OpenGaussCRUDTest', 
            'django_opengauss.tests.OpenGaussQueryTest',
            'django_opengauss.tests.OpenGaussJSONTest',
            'django_opengauss.tests.OpenGaussTransactionTest',
            'django_opengauss.tests.OpenGaussPerformanceTest',
        ]
        
        failures = test_runner.run_tests(test_labels)
        
        # 获取输出
        output = mystdout.getvalue()
        
        return failures, output
        
    finally:
        sys.stdout = old_stdout


if __name__ == '__main__':
    # 如果直接运行此文件，执行测试
    failures, output = run_opengauss_tests()
    print(output)
    
    if failures:
        print(f"\n测试失败: {failures} 个测试用例未通过")
        sys.exit(1)
    else:
        print("\n所有测试通过! ✅")
        sys.exit(0)